package org.pentaho.kettle.plugin.mqtt.steps.consumer;

import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.pentaho.di.core.row.RowDataUtil;

import java.util.Date;
import java.util.concurrent.TimeUnit;


public class MqttClientSubscriber implements MqttCallback,MqttCallbackExtended  {

    private MqttClient client;
    private MqttConnectOptions conOpt;
    private ConsumerStepMeta meta;
    private ConsumerStepData data;
    private ConsumerStep step;

    String topic;
    String broker;
    String clientId;

    String username;
    String password;

    int qos = 1;
    boolean cleanSession = true;
    Object[] r;


    public MqttClientSubscriber(ConsumerStepMeta smi, ConsumerStepData sdi, ConsumerStep step, Object[] r) {
        this.meta = smi;
        this.data = sdi;
        this.step = step;
        this.r = r;
        topic = smi.getSubTopic();
        broker = smi.getSubBroker();
        clientId = String.valueOf(System.currentTimeMillis());
        username = smi.getUsername();
        password = smi.getPassword();
        qos = smi.getQos();
    }

    public void start() {
        try {
            String tmpDir = System.getProperty("java.io.tmpdir");
            MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
             if (conOpt==null){
                   conOpt = new MqttConnectOptions();
             }
            if (StringUtils.isNotBlank(meta.getUsername()) && StringUtils.isNotBlank(meta.getPassword())) {
                conOpt.setUserName(meta.getUsername());
                conOpt.setPassword(meta.getPassword().toCharArray());
            }
            conOpt.setAutomaticReconnect(true);
            conOpt.setConnectionTimeout(50000);
            conOpt.setKeepAliveInterval(100);
            //是否清空session，设置为false表示服务器会保留客户端的连接记录（订阅主题，qos），客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
            //设置为true表示每次连接到服务端都是以新的身份
            conOpt.setCleanSession(cleanSession);
            if (client == null) {
                client = new MqttClient(broker, clientId);
            }
            if (!client.isConnected()) {
                client.connect(conOpt);
                client.setCallback(this);
                client.subscribe(topic, qos);
                step.logBasic("Connected success...");
            }else {//这里的逻辑是如果连接成功就重新连接
                stop();
                client.setCallback(this);
                client.subscribe(topic, qos);
                step.logBasic("reConnected success...");
            }
        } catch (MqttException me) {
            step.logError(me.getMessage());
            step.logError(me.getCause().toString());
            me.printStackTrace();
        }
    }

    public void stop() {
        try {
            client.unsubscribe(topic);
            client.disconnect();
            client.close();
        } catch (MqttException me) {
            step.logBasic("reason " + me.getReasonCode());
            step.logBasic("msg " + me.getMessage());
            step.logBasic("loc " + me.getLocalizedMessage());
            step.logBasic("cause " + me.getCause());
            step.logBasic("excep " + me);
            me.printStackTrace();
        }
    }

   public synchronized void reConnect() {
       if(client.isConnected()){
           return;
       }
       int times=0;
       while (!client.isConnected()) {
           // 每隔5秒重试一次
           try {
               TimeUnit.SECONDS.sleep(10);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
           if (times==10){
               break;
           }
           if (null != client) {
               if (!client.isConnected()) {
                   if (null != conOpt) {
                       try {
                           client.reconnect();
                           //client.setCallback(this);
                           if (client.isConnected()) {
                               client.subscribe(topic, qos);
                               step.logBasic("重连成功" + topic);
                               times++;
                               return;
                           }
                       } catch (MqttException e) {
                           e.printStackTrace();
                       }
                   } else {
                       step.logBasic("mqttConnectOptions is null");
                       break;
                   }
               }
           }
       }
    }

    public void connectionLost(Throwable throwable) {
        System.out.println("失去连接:" + throwable.getMessage());
        //reConnect();
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String time = new Date().toString();
        String messageStr = new String(message.getPayload());
        String messageQos = new String("" + message.getQos());

        try {
            Object[] outputRow = RowDataUtil.addValueData(r, data.outputRowMeta.size() - 1, messageStr);
            step.putRow(data.outputRowMeta, outputRow);
            outputRow = null;
        } catch (Exception e) {
            e.printStackTrace();
        }

        time = null;
        messageStr = null;
        messageQos = null;
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        // TODO Auto-generated method stub
    }

    @Override
    public void connectComplete(boolean b, String s) {
             step.logBasic("reConnected finish..."+s+"==="+b);
            try {
                if (client!= null) {
                    this.client.subscribe(this.topic, qos);
                    step.logBasic("Connected reSubscribe success...");
                }
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
 }

